package com.atguigu.gmall.realtime.app

import java.text.SimpleDateFormat
import java.util
import java.util.Date

import com.alibaba.fastjson.serializer.SerializeConfig
import com.alibaba.fastjson.{JSON, JSONObject}
import com.atguigu.gmall.realtime.bean.{OrderDetail, OrderInfo, OrderWide}
import com.atguigu.gmall.realtime.util.{MyEsUtil, MyKafkaUtil, OffsetManager, RedisUtil}
import org.apache.commons.lang3.time.DateUtils
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

import scala.collection.JavaConverters._
import scala.collection.mutable.ListBuffer

object OrderWideApp {


  // 1  环境
  // 2  读取redis的偏移量(2个)
  // 3  根据偏移量加载kafka流 (2个)
  // 4  从 kafka流提取 偏移量 (2个)
  // 5  分别转换两个流为结构化对象 (2个)
  // 6  分别对需要的维度数据进行查询 （2个）
  // 7  双流join
  // 8  把数据写入到ES中
  // 9  提交偏移量 (2个)

  def main(args: Array[String]): Unit = {
    // 1  环境
    val sparkConf: SparkConf = new SparkConf().setAppName("order_wide_app").setMaster("local[4]")
    val ssc = new StreamingContext(sparkConf, Seconds(5))
    val orderInfoTopic = "DWD_ORDER_INFO_I"
    val orderDetailTopic = "DWD_ORDER_DETAIL_I"
    val groupId = "order_wide_group"
    // 2  读取redis的偏移量(2个)
    val orderInfoOffsetMap: Map[TopicPartition, Long] = OffsetManager.getOffset(orderInfoTopic,groupId)
    val orderDetailOffsetMap: Map[TopicPartition, Long] = OffsetManager.getOffset(orderDetailTopic,groupId)

    // 3  根据偏移量加载kafka流 (2个)
    //order_info
    var orderInfoInputDstream: InputDStream[ConsumerRecord[String, String]]=null
    if(orderInfoOffsetMap!=null&&orderInfoOffsetMap.size>0){
      orderInfoInputDstream= MyKafkaUtil.getKafkaStream(orderInfoTopic,ssc,orderInfoOffsetMap,groupId)
    }else{
      orderInfoInputDstream= MyKafkaUtil.getKafkaStream(orderInfoTopic,ssc,groupId)
    }
    //order_detail
    var orderDetailInputDstream: InputDStream[ConsumerRecord[String, String]]=null
    if(orderDetailOffsetMap!=null&&orderDetailOffsetMap.size>0){
      orderDetailInputDstream= MyKafkaUtil.getKafkaStream(orderDetailTopic,ssc,orderDetailOffsetMap,groupId)
    }else{
      orderDetailInputDstream= MyKafkaUtil.getKafkaStream(orderDetailTopic,ssc,groupId)
    }


    // 4  从 kafka流提取 偏移量 (2个)
    //order_info的偏移量
    var orderInfoOffsetRanges: Array[OffsetRange]=null
    val orderInfoInputDstreamAfterOffset: DStream[ConsumerRecord[String, String]] = orderInfoInputDstream.transform { rdd =>
      orderInfoOffsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
      rdd
    }
    //order_detail的偏移量
    var orderDetailOffsetRanges: Array[OffsetRange]=null
    val orderDetailInputDstreamAfterOffset: DStream[ConsumerRecord[String, String]] = orderDetailInputDstream.transform { rdd =>
      orderDetailOffsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
      rdd
    }
    // 5  分别转换两个流为结构化对象 (2个) 通用 map jsonobject 专用 case class  bean
    //  OrderInfo    OrderDetail  case class
    val orderInfoDstream: DStream[OrderInfo] = orderInfoInputDstreamAfterOffset.map { record =>
      val jsonString: String = record.value()
      val orderInfo: OrderInfo = JSON.parseObject(jsonString, classOf[OrderInfo])
      //补充时间字段
      val createDateTimeArr: Array[String] = orderInfo.create_time.split(" ")
      orderInfo.create_date = createDateTimeArr(0)
      orderInfo.create_hour = createDateTimeArr(1).split(":")(0)
      orderInfo
    }


    val orderDetailDstream: DStream[OrderDetail] = orderDetailInputDstreamAfterOffset.map { record =>
      val jsonString: String = record.value()
      val orderDetail: OrderDetail = JSON.parseObject(jsonString, classOf[OrderDetail])
      orderDetail
    }


    // 6  分别对需要的维度数据进行查询 （2个）
    val orderInfoWithDimDstream: DStream[OrderInfo] = orderInfoDstream.mapPartitions { orderInfoItr =>
      val jedisClient: Jedis = RedisUtil.getJedisClient
      val orderInfoList: List[OrderInfo] = orderInfoItr.toList
      for (orderInfo <- orderInfoList) {
        //补充维度信息 ： 地区
        val provinceKey = "base_province:" + orderInfo.province_id
        val provinceJson: String = jedisClient.get(provinceKey)
        val provinceJsonObj: JSONObject = JSON.parseObject(provinceJson)
        orderInfo.province_name = provinceJsonObj.getString("name")
        orderInfo.province_area_code = provinceJsonObj.getString("area_code") // ali datav  quickbi baidu suger
        orderInfo.province_3166_2_code = provinceJsonObj.getString("iso_3166_2") // kibana
        orderInfo.province_iso_code = provinceJsonObj.getString("iso_code") // superset

        // 补充维度信息: 用户基本信息  :年龄 性别
        val userInfoKey = "user_info:" + orderInfo.user_id
        val userInfoJson: String = jedisClient.get(userInfoKey)
        val userInfoJsonObj: JSONObject = JSON.parseObject(userInfoJson)
        orderInfo.user_gender = userInfoJsonObj.getString("gender")
        val birthday = userInfoJsonObj.getString("birthday")
        val simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd")
        val birthdayDate: Date = simpleDateFormat.parse(birthday)
        val nowDate = new Date()
        val age: Long = (nowDate.getTime - birthdayDate.getTime) / 1000 / 60 / 60 / 24 / 365
        orderInfo.user_age = age.toInt
      }
      jedisClient.close()
      orderInfoList.toIterator
    }

   // orderInfoWithDimDstream.print(1000)


    // 7  双流join
    val orderInfoWithIdDstream: DStream[(Long, OrderInfo)] = orderInfoWithDimDstream.map(orderInfo =>(orderInfo.id,orderInfo))
    val orderDetailWithOidDstream: DStream[(Long, OrderDetail)] = orderDetailDstream.map(orderDetail=>(orderDetail.order_id,orderDetail))

    val orderFulljoinedDStream: DStream[(Long, (Option[OrderInfo], Option[OrderDetail]))] = orderInfoWithIdDstream.fullOuterJoin(orderDetailWithOidDstream)

    //join本批次的同时，还有查对面流的缓存
    val orderWideDStream: DStream[OrderWide] = orderFulljoinedDStream.flatMap { case (orderId, (orderInfoOption, orderDetailOption)) =>
      val orderWideList = new ListBuffer[OrderWide]()
      val jedisClient: Jedis = RedisUtil.getJedisClient
      if (orderInfoOption != None) {
        val orderInfo: OrderInfo = orderInfoOption.get
        //主表有值
        //1  主表有值且从表有值  直接产生join结果 => 形成一行宽表数据=> 存放到一个集合中
        if (orderDetailOption != None) {
          val orderDetail: OrderDetail = orderDetailOption.get
          val orderWide = new OrderWide(orderInfo, orderDetail)
          orderWideList.append(orderWide)
        }
        //2  主表 写缓存
        //  type ?  string  通过key 一对一查询    key ?   order_wide:order_info:[102]  value?   orderInfoJson
        // field/score? 无    写入api?  setex   读取api？  get  过期时效？ 2个小时
        val orderInfoKey = s"order_wide:order_info:${orderInfo.id}"
        val orderInfoJson: String = JSON.toJSONString(orderInfo, new SerializeConfig(true))
        jedisClient.setex(orderInfoKey, 3600 * 2, orderInfoJson)
        //3  主表 读缓存   产生join结果    => 形成一行或多行宽表数据=> 存放到一个集合中
        val orderDetailKey = s"order_wide:order_detail:${orderInfo.id}"
        val orderDetailJsonSet: util.Set[String] = jedisClient.smembers(orderDetailKey)
        for (orderDetailJson <- orderDetailJsonSet.asScala) {
          val orderDetail: OrderDetail = JSON.parseObject(orderDetailJson, classOf[OrderDetail])
          val orderWide = new OrderWide(orderInfo, orderDetail)
          orderWideList.append(orderWide)
        }
      } else {
        //主表不在 从表在
        val orderDetail: OrderDetail = orderDetailOption.get
        //  1  从表要写缓存
        // type?   set    key?  order_wide:order_detail:[order_id]  value ?  orderDetailJsons..多个
        //  field/score?  无  写入api  sadd ?读取api? smembers    过期时间？   2个小时
        val orderDetailKey = s"order_wide:order_detail:${orderDetail.order_id}"
        val orderDetailJson: String = JSON.toJSONString(orderDetail, new SerializeConfig(true))
        jedisClient.sadd(orderDetailKey, orderDetailJson)
        jedisClient.expire(orderDetailKey, 3600 * 2)

        //  2  从表查询主表缓存  产生join结果 => 形成一行宽表数据=> 存放到一个集合中
        val orderInfoKey = s"order_wide:order_info:${orderDetail.order_id}"
        val orderInfoJson: String = jedisClient.get(orderInfoKey)
        val orderInfo: OrderInfo = JSON.parseObject(orderInfoJson, classOf[OrderInfo])
        val orderWide = new OrderWide(orderInfo, orderDetail)
        orderWideList.append(orderWide)
      }
      jedisClient.close()
      orderWideList

    }
    orderWideDStream.print(1000)


    // 8  把数据写入到ES中
    orderWideDStream.foreachRDD{rdd=>
       rdd.foreachPartition{orderWideItr=>
         val orderWideList: List[OrderWide] = orderWideItr.toList
         if(orderWideList.size>0){
           val orderWideFirst: OrderWide = orderWideList(0)
           val indexDate: String = orderWideFirst.create_date
           val indexName=s"gmall0609_order_wide_$indexDate"
           val orderWideWithIdList: List[(String, OrderWide)] = orderWideList.map(orderWide=>(orderWide.detail_id.toString,orderWide))
           MyEsUtil.bulkSave(orderWideWithIdList,indexName)
         }
       }
      // 9  提交偏移量 (2个)
      OffsetManager.saveOffset(orderInfoTopic,groupId,orderInfoOffsetRanges)
      OffsetManager.saveOffset(orderDetailTopic,groupId,orderDetailOffsetRanges)
    }

    ssc.start()
    ssc.awaitTermination()
  }

}
